package socket

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"time"

	"gddgame.cc/galaxy/asteroid"
	"gddgame.cc/galaxy/asteroid/strategy/hash"
	"gddgame.cc/galaxy/binocle/logic"
	"gddgame.cc/galaxy/utils"
	"gddgame.cc/galaxy/utils/cache"
	"gddgame.cc/galaxy/utils/def"
	"gddgame.cc/galaxy/utils/serialize"
)

type TargetNodeHandler func(target string, node string, number int)

type AgentSocket interface {
	ListenStream(event string, fn ListenStreamHandler)
	ListenDo(event string, fn ListenEventHandler)
	Request(target string) Message
	RequestService(target string, region string) Message
	OnTargetConnect(target string, handler TargetNodeHandler)
}

type AppSocket interface {
	ListenStream(event string, fn ListenStreamHandler)
	ListenDo(event string, fn ListenEventHandler)
	Request(target string) Message
	OnTargetConnect(target string, handler TargetNodeHandler)
}

type ServiceSocket interface {
	ListenStream(event string, fn ListenStreamHandler)
	ListenDo(event string, fn ListenEventHandler)
	Request(target string) Message
	RequestService(target string, region string) Message
}

type socket struct {
	def.Logger
	logic.AgentInf
	channel asteroid.PublishPipe

	name string

	reqIdMap    sync.Map
	reqRespPool sync.Pool

	no      uint64
	timeout time.Duration

	cache  *cache.MemoryCache
	stores sync.Map

	onTargetMutex sync.RWMutex
	onTarget      map[string][]TargetNodeHandler

	onStream map[string]ListenStreamHandler
	onEvent  map[string]ListenEventHandler
}

func NewSocket(name string, logger def.Logger, agent logic.AgentInf, subscriber asteroid.Sub, channel asteroid.PublishPipe) *socket {
	s := &socket{
		Logger:   logger,
		AgentInf: agent,
		channel:  channel,
		name:     name,

		reqIdMap:    sync.Map{},
		reqRespPool: sync.Pool{},

		no:      0,
		timeout: 2 * time.Second,

		cache:  cache.Memory(logger),
		stores: sync.Map{},

		onTargetMutex: sync.RWMutex{},
		onTarget:      map[string][]TargetNodeHandler{},

		onStream: make(map[string]ListenStreamHandler),
		onEvent:  map[string]ListenEventHandler{},
	}
	s.reqRespPool.New = func() interface{} {
		message := &reqResp{
			socket: s,
		}
		return message
	}
	subscriber.On("socket", func(message asteroid.Message) {
		msg := newMsg()
		message.UnSerialize(msg)
		switch msg.Type {
		case RequestOnceType:
			if handler, ok := s.onEvent[msg.Event]; ok {
				request := s.request(msg.Id, message.GetOrigin(), message.GetFrom(), message.GetTarget())
				slot := serialize.GetSlot()
				request.slot = slot
				_ = request.slot.Unmarshal(msg.Payload)
				handler(request)
				if !request.manual {
					// todo first miss
					request.Response(true)
				}
			}
		case RequestStream:
			iid := message.GetFrom() + ":" + msg.Id
			if handler, ok := s.onStream[msg.Event]; ok {
				request := s.request(msg.Id, message.GetOrigin(), message.GetFrom(), message.GetTarget())
				slot := serialize.GetSlot()
				request.slot = slot
				_ = request.slot.Unmarshal(msg.Payload)
				result := handler(request)
				if result {
					s.reqIdMap.Store(iid, request)
				}
				message.Response(result)
			}
		case FinishStream:
			iid := message.GetFrom() + ":" + msg.Id
			if q, ok := s.reqIdMap.Load(iid); ok {
				request := q.(*reqResp)
				s.reqIdMap.Delete(iid)
				request.release()
			}
		case ResponseOnceType:
			iid := msg.Id
			q, ok := s.reqIdMap.Load(iid)
			if ok {
				request := q.(*reqResp)
				request.time.Stop()
				slot := serialize.GetSlot()
				request.slot = slot
				_ = request.slot.Unmarshal(msg.Payload)
				request.handler(request, nil)
				s.reqIdMap.Delete(iid)
				request.Release()
			}
		case ResponseErrorType:
			iid := msg.Id
			q, ok := s.reqIdMap.Load(iid)
			if ok {
				request := q.(*reqResp)
				request.time.Stop()
				request.handler(nil, errors.New(msg.Event))
				s.reqIdMap.Delete(iid)
				request.release()
			}
		case ResponseStream:
			iid := msg.Id
			q, ok := s.reqIdMap.Load(iid)
			if ok {
				request := q.(*reqResp)
				slot := serialize.GetSlot()
				request.slot = slot
				_ = request.slot.Unmarshal(msg.Payload)
				request.receive <- request
			}
		}
		msg.Release()
	})
	return s
}

func (socket *socket) context() context.Context {
	return context.Background()
}

func (socket *socket) publish(agent string) *asteroid.Publisher {
	publisher, _ := socket.channel.Publish(agent)
	return publisher
}

func (socket *socket) OnTargetConnect(target string, handler TargetNodeHandler) {
	target = socket.FullName(target)
	socket.onTargetMutex.Lock()
	fns, ok := socket.onTarget[target]
	if !ok {
		fns = []TargetNodeHandler{}
	}
	socket.onTarget[target] = append(fns, handler)
	socket.onTargetMutex.Unlock()
	if !ok {
		publisher := socket.publish(target)

		publisher.SetStrategy(hash.NewStrategy())
		publisher.OnJoin(func(node string) {
			socket.onTargetMutex.RLock()
			for _, fn := range socket.onTarget[target] {
				fn(target, node, len(publisher.AllNode()))
			}
			socket.onTargetMutex.RUnlock()
		})
		publisher.OnReconnect(func(node string) {
			socket.onTargetMutex.RLock()
			for _, fn := range socket.onTarget[target] {
				fn(target, node, len(publisher.AllNode()))
			}
			socket.onTargetMutex.RUnlock()
		})
	}
}

func (socket *socket) Request(target string) Message {
	request := socket.reqRespPool.Get().(*reqResp)
	if request.slot == nil {
		request.slot = serialize.GetSlot()
	}
	request.from = socket.FullName(target)
	request.region = socket.name
	return request
}

func (socket *socket) RequestService(target string, from string) Message {
	request := socket.reqRespPool.Get().(*reqResp)
	if request.slot == nil {
		request.slot = serialize.GetSlot()
	}
	request.from = socket.FullName(target)
	request.region = socket.FullName(from)
	return request
}

func (socket *socket) ListenStream(event string, fn ListenStreamHandler) {
	socket.onStream[event] = fn
}

func (socket *socket) ListenDo(event string, fn ListenEventHandler) {
	socket.onEvent[event] = fn
}

func (socket *socket) request(id string, from string, node string, region string) *reqResp {
	request := socket.reqRespPool.Get().(*reqResp)
	request.id = id
	// 上层主体对象
	request.from = from
	//上层主体对象，或者主体对象下的特定服务
	request.region = region
	request.node = node
	request.Logger = socket

	if _, ok := socket.stores.Load(region); !ok {
		socket.stores.Store(region, socket.cache.Store(region))
	}
	store, _ := socket.stores.Load(region)
	request.Store = store.(def.Store)
	return request
}

func (socket *socket) release(message *reqResp) {
	socket.reqRespPool.Put(message)
}

func (socket *socket) responseError(message *reqResp, id string, err error) {
	msg := newMsg()
	msg.Type = ResponseErrorType
	msg.Id = id
	msg.Event = err.Error()
	socket.channel.Pipe(message.from).Node(message.node).OriginId(socket.name).TargetId(message.region).Data(msg).Send("socket")
}

func (socket *socket) responseDo(message *reqResp, id string, data def.OptimizeSerializer, serializer []def.AutoSerializer) {
	msg := newMsg()
	msg.Type = ResponseOnceType
	msg.Id = id
	msg.Payload, _ = data.Marshal()
	socket.channel.Pipe(message.from).Node(message.node).OriginId(socket.name).TargetId(message.region).Data(msg).AutoRelease(data).AutoRelease(serializer...).Send("socket")
}

func (socket *socket) responseStream(message *reqResp, id string, data def.OptimizeSerializer, serializer []def.AutoSerializer) {
	msg := newMsg()
	msg.Type = ResponseStream
	msg.Id = id
	msg.Payload, _ = data.Marshal()
	socket.channel.Pipe(message.from).Node(message.node).OriginId(socket.name).TargetId(message.region).Data(msg).AutoRelease(data).AutoRelease(serializer...).Send("socket")
}

func (socket *socket) finishStream(message *reqResp, id string) {
	msg := newMsg()
	msg.Type = FinishStream
	msg.Id = id
	if message.node != "" {
		socket.channel.Pipe(message.from).Node(message.node).OriginId(socket.name).TargetId(message.region).Data(msg).Send("socket")
	} else {
		socket.publish(message.from).All().OriginId(socket.name).TargetId(message.region).Data(msg).Send("socket")
	}
	socket.reqIdMap.Delete(id)
	message.release()
}

func (socket *socket) message(message *reqResp, messageType MessageType, event string, data def.OptimizeSerializer, serializer []def.AutoSerializer) string {
	id := utils.ToString(atomic.AddUint64(&socket.no, 1))
	publisher := socket.publish(message.from)
	var pipe *asteroid.Pipe
	if message.strategy != nil {
		message.strategy.SetNodes(publisher.AllNode())
		pipe = publisher.Direct(message.strategy.Alloc(id))
	} else if message.hash {
		pipe = publisher.Auto()
	} else {
		pipe = publisher.All()
	}

	msg := newMsg()
	msg.Type = messageType
	msg.Id = id
	msg.Event = event
	msg.Payload, _ = data.Marshal()

	pipe.OriginId(socket.name).TargetId(message.region).Data(msg).AutoRelease(data).AutoRelease(serializer...).Send("socket")
	if message.handler != nil {
		socket.reqIdMap.Store(id, message)
	}
	return id
}
